package com.ydl.learning.flink.demo.myFunction

import com.ydl.learning.flink.demo.entity.TestEntity
import org.apache.flink.streaming.api.scala.async.{ResultFuture, RichAsyncFunction}
import org.slf4j.{Logger, LoggerFactory}

/**
 * 异步function
 *
 * @author ydl
 * @since 2021/4/14
 */
class AsyncFuncion extends RichAsyncFunction[TestEntity, TestEntity] {
  private val log: Logger = LoggerFactory.getLogger(classOf[AsyncFuncion])

  override def asyncInvoke(input: TestEntity, resultFuture: ResultFuture[TestEntity]): Unit = {
    log.info("set name")
    input.setName("ydl")
    Thread.sleep(5000)
    resultFuture.complete(Set(input))
  }

  override def timeout(input: TestEntity, resultFuture: ResultFuture[TestEntity]): Unit = {
    log.error("timeout:" + input)
  }
}
